﻿using BeetleX.Dispatchs;
using BeetleX.EventArgs;
using BeetleX.XRPC.Awaiter;
using BeetleX.XRPC.Clients;
using BeetleX.XRPC.Events;
using BeetleX.XRPC.Packets;
using EventNext;
using EventNext.Events;
using Newtonsoft.Json;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Reflection;
using System.Runtime;
using System.Threading;
using System.Threading.Tasks;

#nullable disable
namespace BeetleX.XRPC
{
    public class XRPCServer : ServerHandlerBase, IDisposable
    {
        public const string NAME_PROPERTY_TAG = "__XRPC_NET_USER";
        private long mRequests;
        private long mResponses;
        //private IServer mServer;
        private FileLogWriter mFileLog;
        private DispatchCenter<RPCPacket> mRequestDispatchCenter;
        private Dictionary<string, object> mProperties = new Dictionary<string, object>((IEqualityComparer<string>)StringComparer.OrdinalIgnoreCase);
        private AwaiterFactory AwaiterFactory = new AwaiterFactory(true);
        public EventHandler<ServerLogEventArgs> Logger;
        private EventCenter mEventCenter = new EventCenter();
        private int mEventCenterInit;
        private ConcurrentDictionary<string, DelegatePublisher> mDelegatePublishers = new ConcurrentDictionary<string, DelegatePublisher>((IEqualityComparer<string>)StringComparer.OrdinalIgnoreCase);
        private ConcurrentDictionary<string, DelegateHandler> mDelegateHandlers = new ConcurrentDictionary<string, DelegateHandler>((IEqualityComparer<string>)StringComparer.OrdinalIgnoreCase);

        public XRPCServer()
        {
            this.mFileLog = new FileLogWriter("BEETLEX_XRPC_SERVER");
            this.mRequestDispatchCenter = new DispatchCenter<RPCPacket>(new Action<RPCPacket>(this.OnRequestProcess));
            this.ServerOptions.BufferSize = 8192;
        }

        private void BindRequestParameter(RPCPacket request, EventActionHandler handler)
        {
            request.LoadParameters(handler?.ParametersType);
        }

        private void OnEventNext(RPCPacket e)
        {
            EventInput input = new EventInput();
            input.ID = e.ID;
            input.EventPath = e.Url;
            input.Properties = e.Header;
            input.Data = e.Data;
            if (this.EnableLog(BeetleX.EventArgs.LogType.Debug))
            {
                string str = JsonConvert.SerializeObject(input);
                this.Log(BeetleX.EventArgs.LogType.Debug, string.Format("[{0}]{1} receive event data:{2}", input.ID, e.Session.RemoteEndPoint, str));
            }
            input.Token = new XRPCEventToken() { Request = e, Server = this, Session = e.Session };
            this.EventCenter.Execute(input, new XRPCServer.EventCompleted() { Server = this, Request = e });
        }

        private void OnRequestProcess(RPCPacket e)
        {
            RPCPacket rpcPacket = e;
            try
            {
                EventPacketProcessingArgs packetProcessingArgs = this.OnProcessing(e);
                if (packetProcessingArgs != null && packetProcessingArgs.Cancel)
                    return;
                if (this.Security && e.Session.Authentication < AuthenticationType.User && e.Url != "/__login/")
                {
                    e.ReplyError((short)500, "No permission operation in secure mode!");
                }
                else
                {
                    EventActionHandler actionHandler = this.EventCenter.GetActionHandler(e.Url);
                    if (actionHandler != null)
                    {
                        this.BindRequestParameter(e, actionHandler);
                        this.OnEventNext(e);
                    }
                    else if (e.Url == "/__System/Ping")
                    {
                        using RPCPacket response = new RPCPacket();
                        if (this.EnableLog(BeetleX.EventArgs.LogType.Debug))
                            this.Log(BeetleX.EventArgs.LogType.Debug, string.Format("[{0}]{1} request {2}", (object)e.ID, (object)e.Session.RemoteEndPoint, (object)e.Url));
                        response.Status = (short)200;
                        this.OnResponse(e, response);
                    }
                    else if (e.Url == "/__login/")
                    {
                        EventLoginArgs e1 = new EventLoginArgs(this, e.Session);
                        e.LoadParameters<string, string>();
                        e1.UserName = (string)e.Data[0];
                        e1.Password = (string)e.Data[1];
                        EventHandler<EventLoginArgs> login = this.Login;
                        if (login != null)
                            login((object)this, e1);
                        if (e1.Success)
                        {
                            e.Session["__XRPC_NET_USER"] = (object)e1.UserName;
                            e.ReplySuccess();
                        }
                        else
                        {
                            string message = e1.Message;
                            if (string.IsNullOrEmpty(message))
                                message = "Invalid username or password!";
                            e.ReplyError(message);
                        }
                    }
                    else
                    {
                        AwaiterItem awaiterItem = this.AwaiterFactory.GetItem(rpcPacket.ID);
                        if (awaiterItem != null)
                        {
                            rpcPacket.ResultType = awaiterItem.ResultType;
                            try
                            {
                                if (rpcPacket.ResultType != null)
                                    rpcPacket.LoadParameters(rpcPacket.ResultType);
                            }
                            catch (Exception ex)
                            {
                                rpcPacket.Status = (short)500;
                                rpcPacket.Data = new object[1] { (ex.Message + "@" + ex.StackTrace) };
                            }
                            this.AwaiterFactory.Completed(awaiterItem, rpcPacket);
                        }
                        else if (rpcPacket.Url.IndexOf("/__delegate/", StringComparison.OrdinalIgnoreCase) == 0)
                            this.InvokeDelegate(rpcPacket);
                        else if (rpcPacket.Url.IndexOf("/__subscribe/", StringComparison.OrdinalIgnoreCase) == 0)
                            this.OnSubscribe(rpcPacket);
                        else if (this.NotFound == null)
                        {
                            if (this.EnableLog(BeetleX.EventArgs.LogType.Debug))
                                this.Log(BeetleX.EventArgs.LogType.Debug, string.Format("[{0}]{1} request {2} not found!", (object)e.ID, (object)e.Session.RemoteEndPoint, (object)e.Url));
                            this.OnResponse(e, new RPCPacket()
                            {
                                Status = (short)404,
                                Data = new object[1] { ("request " + e.Url + " not found!") }
                            });
                        }
                        else
                            this.NotFound((object)this, new EventPacketArgs(this, e));
                    }
                }
            }
            catch (Exception ex)
            {
                if (!this.EnableLog(BeetleX.EventArgs.LogType.Error))
                    return;
                this.Log(BeetleX.EventArgs.LogType.Error, string.Format("[{0}]{1} process {2} error {3}@{4}!", (object)e.ID, (object)e.Session.RemoteEndPoint, (object)e.Url, (object)ex.Message, (object)ex.StackTrace));
            }
        }

        protected EventPacketProcessingArgs OnProcessing(RPCPacket packet)
        {
            if (this.Processing != null)
            {
                try
                {
                    EventPacketProcessingArgs e = new EventPacketProcessingArgs(this, packet);
                    this.Processing((object)this, e);
                    return e;
                }
                catch (Exception ex)
                {
                    if (this.EnableLog(BeetleX.EventArgs.LogType.Error))
                        this.Log(BeetleX.EventArgs.LogType.Error, string.Format("[{0}]{1} processing {2} event error {3}@{4}!", (object)packet.ID, (object)packet.Session.RemoteEndPoint, (object)packet.Url, (object)ex.Message, (object)ex.StackTrace));
                }
            }
            return (EventPacketProcessingArgs)null;
        }

        public int ClientTimeout { get; set; } = 10000;

        public EventHandler<EventLoginArgs> Login { get; set; }

        public event EventHandler<EventPacketProcessingArgs> Processing;

        public event EventHandler<EventPacketArgs> NotFound;

        public void Send(RPCPacket response, ISession[] sessions)
        {
            if (sessions == null)
                return;
            foreach (ISession session in sessions)
            {
                Interlocked.Add(ref this.mResponses, (long)sessions.Length);
                this.Server.Send((object)response, sessions);
            }
        }

        public void Reply(RPCPacket response, RPCPacket request) => this.OnResponse(request, response);

        public Task<RPCPacket> SendWait(RPCPacket request, ISession session)
        {
            return this.SendWait(request, session, (Type[])null);
        }

        internal Task<RPCPacket> SendWait(RPCPacket request, ISession session, Type[] resultTypes)
        {
            (int, TaskCompletionSource<RPCPacket>) tuple = this.AwaiterFactory.Create(request, resultTypes, this.ClientTimeout);
            request.ID = tuple.Item1;
            session.Send((object)request);
            Interlocked.Increment(ref this.mRequests);
            return tuple.Item2.Task;
        }

        public long Requests => this.mRequests;

        public long Responses => this.mResponses;

        public bool Security { get; set; }

        public void Register(object controller) => this.EventCenter.Register(controller);

        public void Register(params Assembly[] assemblies) => this.EventCenter.Register(assemblies);

        public ServerOptions ServerOptions { get; private set; } = new ServerOptions();

        public Options RPCOptions { get; private set; } = new Options();

        public event EventHandler<ConnectedEventArgs> RPCConnected;

        public event EventHandler<SessionEventArgs> RPCDisconnect;

        public void Log(BeetleX.EventArgs.LogType type, string message, params object[] parameters)
        {
            this.Log(type, string.Format(message, parameters));
        }

        public EventCenter EventCenter
        {
            get
            {
                if (Interlocked.CompareExchange(ref this.mEventCenterInit, 1, 0) == 0)
                {
                    this.mEventCenter.LogLevel = (EventNext.LogType)this.ServerOptions.LogLevel;
                    this.mEventCenter.LogOutput += new EventHandler<EventLogArgs>(this.OnEventLog);
                }
                return this.mEventCenter;
            }
        }

        public override void Log(IServer server, ServerLogEventArgs e)
        {
            if (this.Logger == null)
            {
                if (this.RPCOptions.LogToConsole)
                    base.Log(server, e);
                if (!this.RPCOptions.LogToFile)
                    return;
                this.mFileLog.Add(e.Type, e.Message);
            }
            else
                this.Logger((object)server, e);
        }

        public bool EnableLog(BeetleX.EventArgs.LogType logType)
        {
            return this.ServerOptions.LogLevel <= logType;
        }

        public void Log(BeetleX.EventArgs.LogType type, string message)
        {
            try { this.Log((IServer)null, new ServerLogEventArgs(message, type)); } catch { }
        }

        //public override void Connecting(IServer server, ConnectingEventArgs e)
        //{
        //    if (server.Count <= this.ServerOptions.MaxConnections)
        //        return;
        //    e.Cancel = true;
        //    if (!this.EnableLog(BeetleX.EventArgs.LogType.Warring))
        //        return;
        //    this.Log(BeetleX.EventArgs.LogType.Warring, string.Format("XRPC Maximum online limit,{0} closed!", (object)e.Socket.RemoteEndPoint));
        //}

        public override void Connected(IServer server, ConnectedEventArgs e)
        {
            if (RPCOptions.MessageMaxLength> server.Count) { }
            base.Connected(server, e);
            this.RPCConnected?.Invoke(this, e);
        }

        public override void Disconnect(IServer server, SessionEventArgs e)
        {
            base.Disconnect(server, e);
            this.RPCDisconnect?.Invoke(this, e);
        }

        protected override void OnReceiveMessage(IServer server, ISession session, object message)
        {
            base.OnReceiveMessage(server, session, message);
            RPCPacket data = (RPCPacket)message;
            if (this.EnableLog(BeetleX.EventArgs.LogType.Debug))
                this.Log(BeetleX.EventArgs.LogType.Debug, string.Format("[{0}]{1} receive message {2}@{3}", (object)data.ID, (object)session.RemoteEndPoint, (object)data.Url, (object)data.ID));
            Interlocked.Increment(ref this.mRequests);
            this.Server.UpdateSession(session);
            this.mRequestDispatchCenter.Enqueue(data, 1);
        }

        public void Open()
        {
            ServerPacket serverPacket = new ServerPacket();
            serverPacket.Options = this.RPCOptions;
            if (this.ServerOptions.BufferSize < 16384)
                this.ServerOptions.BufferSize = 16384;
            this.Server = SocketFactory.CreateTcpServer((IServerHandler)this, (IPacket)serverPacket, this.ServerOptions);
            this.Server.WriteLogo = new Action(this.OutputLogo);
            this.Server.Open();
        }

        private void OutputLogo()
        {
            string str = "\r\n" + " -----------------------------------------------------------------------------\r\n" + "          ____                  _     _         __   __\r\n         |  _ \\                | |   | |        \\ \\ / /\r\n         | |_) |   ___    ___  | |_  | |   ___   \\ V / \r\n         |  _ <   / _ \\  / _ \\ | __| | |  / _ \\   > <  \r\n         | |_) | |  __/ |  __/ | |_  | | |  __/  / . \\ \r\n         |____/   \\___|  \\___|  \\__| |_|  \\___| /_/ \\_\\ \r\n\r\n                                           xrpc framework   \r\n\r\n" + " -----------------------------------------------------------------------------\r\n" + "\r\n" + string.Format(" ServerGC    [{0}]\r\n", GCSettings.IsServerGC) + string.Format(" BeetleX     Version [{0}]\r\n", typeof(BXException).Assembly.GetName().Version) + string.Format(" EventNext   Version [{0}]\r\n", typeof(NextQueue).Assembly.GetName().Version) + string.Format(" XRPC        Version [{0}] \r\n", typeof(XRPCServer).Assembly.GetName().Version) + " -----------------------------------------------------------------------------\r\n";
            foreach (ListenHandler listen in (IEnumerable<ListenHandler>)this.Server.Options.Listens)
                str += string.Format(" {0}\r\n", (object)listen);
            this.Server.Log(BeetleX.EventArgs.LogType.Info, (ISession)null, str + " -----------------------------------------------------------------------------\r\n");
        }

        private void OnEventLog(object sender, EventLogArgs e) => this.Log((BeetleX.EventArgs.LogType)e.Type, e.Message);

        internal void OnResponse(RPCPacket request, RPCPacket response)
        {
            Interlocked.Increment(ref this.mResponses);
            response.ID = request.ID;
            request.Session.Send((object)response);
        }

        public object this[string name]
        {
            get => mProperties.TryGetValue(name, out object obj) ? obj : null;
            set => mProperties[name] = value;
        }

        public void Dispose()
        {
            if (Server != null)
            {
                this.Server.Dispose();
                this.Server = null;
            }
        }

        public static XRPCEventToken EventToken
        {
            get => (XRPCEventToken)EventCenter.EventActionContext?.Input.Token;
        }

        public T Delegate<T>(ISession session) where T : System.Delegate
        {
            string delegateName = DelegateHandler.GetDelegateName(typeof(T));
            if (!(session[delegateName] is ServerDelegateHandler serverDelegateHandler))
            {
                serverDelegateHandler = new ServerDelegateHandler(typeof(T));
                serverDelegateHandler.Init();
                serverDelegateHandler.Bind(session, this);
                session[delegateName] = (object)serverDelegateHandler;
            }
            return (T)serverDelegateHandler.SessionDelegateProxy;
        }

        public T GetClient<T>(ISession session)
        {
            string name = typeof(T).Name;
            object client1 = session[name];
            if (client1 != null)
                return (T)client1;
            object client2 = DispatchProxy.Create<T, XRPCSeverInvokeClientDispatch>();
            XRPCSeverInvokeClientDispatch invokeClientDispatch = (XRPCSeverInvokeClientDispatch)client2;
            invokeClientDispatch.Session = session;
            invokeClientDispatch.Server = this;
            invokeClientDispatch.Type = typeof(T);
            invokeClientDispatch.InitHandlers();
            return (T)client2;
        }

        private DelegatePublisher GetPublisher(string action)
        {
            DelegatePublisher publisher = new DelegatePublisher(action);
            if (!this.mDelegatePublishers.TryAdd(action, publisher))
                this.mDelegatePublishers.TryGetValue(action, out publisher);
            return publisher;
        }

        private void OnSubscribe(RPCPacket packet)
        {
            string action;
            packet.Url.SubRightWith('/', out action);
            this.GetPublisher(action).Add(packet.Session);
            packet.ReplySuccess();
        }

        public T Publish<T>() where T : System.Delegate
        {
            Type type = typeof(T);
            return (T)this.GetPublisher(DelegateHandler.GetDelegateName(type)).CreateDelegate(type);
        }

        private async void InvokeDelegate(RPCPacket packet)
        {
            XRPCServer xrpcServer = this;
            string action;
            packet.Url.SubRightWith('/', out action);
            DelegateHandler handler;
            if (xrpcServer.mDelegateHandlers.TryGetValue(action, out handler))
            {
                try
                {
                    packet.LoadParameters(handler.Parameters);
                    object result = handler.Delegate.DynamicInvoke(packet.Data);
                    if (!handler.IsVoid)
                    {
                        await (Task)result;
                        if (handler.TargetReturnType != (Type)null)
                        {
                            object obj = handler.GetValue(result);
                            packet.Reply(obj);
                        }
                        else
                            packet.ReplySuccess();
                    }
                    result = (object)null;
                    action = (string)null;
                    handler = (DelegateHandler)null;
                }
                catch (Exception ex)
                {
                    if (!handler.IsVoid)
                        packet.ReplyError((short)500, action + " delegate invoke error " + ex.Message + "!");
                    if (!xrpcServer.EnableLog(BeetleX.EventArgs.LogType.Error))
                    {
                        action = null;
                        handler = null;
                    }
                    else
                    {
                        xrpcServer.Log(BeetleX.EventArgs.LogType.Error, action + " delegate invoke error " + ex.Message + "@" + ex.StackTrace);
                        action = null;
                        handler = null;
                    }
                }
            }
            else if (!packet.NeedReply)
            {
                action = null;
                handler = null;
            }
            else
            {
                packet.ReplyError((short)404, action + " delegate not found!");
                action = null;
                handler = null;
            }
        }

        public XRPCServer AddDelegate<T>(T handler) where T : System.Delegate
        {
            DelegateHandler delegateHandler = new DelegateHandler(typeof(T));
            delegateHandler.Delegate = (System.Delegate)handler;
            delegateHandler.Init();
            this.mDelegateHandlers[delegateHandler.Name] = delegateHandler;
            return this;
        }

        private class EventCompleted : IEventCompleted
        {
            public XRPCServer Server { get; set; }

            public RPCPacket Request { get; set; }

            public void Completed(IEventOutput data)
            {
                using RPCPacket response = new RPCPacket();
                response.Status = (short)data.EventError;
                response.Header = data.Properties;
                response.Data = data.Data;
                data.Token = (object)null;
                if (this.Server.EnableLog(BeetleX.EventArgs.LogType.Debug))
                {
                    string str = JsonConvert.SerializeObject((object)data);
                    this.Server.Log(BeetleX.EventArgs.LogType.Debug, string.Format("[{0}]{1} send event data:{2}", (object)data.ID, (object)this.Request.Session.RemoteEndPoint, (object)str));
                }
                this.Server.OnResponse(this.Request, response);
            }
        }
    }
}
